package simpleim

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/IBM/sarama"
	"github.com/ecodeclub/ekit/syncx"
	"github.com/gorilla/websocket"
	"jk-time/microBook/pkg/saramax"
	"log"
	"net/http"
	"strconv"
	"time"
)

/**
 * (网关)接收信息 -- produce --失败回退 -- consume -- 发送消息
 */
type WsGateway struct {
	// 连接了这个实例的客户端
	// uid作为key
	// 实践中要考虑到不同的设备，
	// 那么这个 key 可能是一个复合结构，例如 uid + 设备
	conns      *syncx.Map[int64, *Conn]
	groups     *syncx.Map[string, []int64]
	svc        *IMService
	client     sarama.Client
	instanceId string
}

func NewWsGateway(conns *syncx.Map[int64, *Conn], svc *IMService, client sarama.Client, instanceId string) *WsGateway {
	return &WsGateway{conns: conns, svc: svc, client: client, instanceId: instanceId}
}

func (ws *WsGateway) Start(addr string) error {
	mux := http.NewServeMux()
	//read,kafka转发给其他结点
	mux.HandleFunc("/ws", ws.wsHandler)

	//kafka消费，再send发送
	err := ws.subscribeMsg()
	if err != nil {
		return err
	}
	return http.ListenAndServe(addr, mux)
}

// listen,发送kafka
func (ws *WsGateway) wsHandler(writer http.ResponseWriter, request *http.Request) {
	upgrader := websocket.Upgrader{
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}
	uid := ws.GetUid(request)
	//group := ws.GetGroup(request)

	conn, err := upgrader.Upgrade(writer, request, nil)
	if err != nil {
		writer.Write([]byte("初始化 websocket 失败"))
		return
	}

	c := &Conn{conn}
	ws.conns.Store(uid, c)
	//val,ok := ws.groups.Load(group)
	//if !ok{
	//	ws.groups.Store(group, make([]int64, 0,1))
	//}
	//val = append(val,uid)

	go func() {
		for {
			_, message, err := conn.ReadMessage()
			if err != nil {
				//去掉结点
				ws.conns.Delete(uid)
				fmt.Println("Error reading message:", err)
				break
			}
			// 转发到后端
			var msg Message
			err = json.Unmarshal(message, &msg)
			if err != nil {
				log.Println("非法数据格式", err)
				continue
			}
			// 通知后端
			go func() {
				ctx, cancel := context.WithTimeout(context.Background(),
					time.Second*3)
				defer cancel()

				err1 := ws.svc.Produce(ctx, uid, msg)
				if err1 != nil {
					err1 = c.Send(Message{Seq: msg.Seq, Type: "result", Content: "FAILED"})
					if err1 != nil {
						log.Println(err1)
					}
				}
			}()
		}
	}()
}
func (ws *WsGateway) GetUid(req *http.Request) int64 {
	uidStr := req.Header.Get("uid")
	uid, _ := strconv.ParseInt(uidStr, 10, 64)
	return uid
}
func (ws *WsGateway) GetGroup(req *http.Request) string {
	groupStr := req.Header.Get("group")
	return groupStr
}
func (ws *WsGateway) subscribeMsg() error {
	cg, err := sarama.NewConsumerGroupFromClient(ws.instanceId, ws.client)
	if err != nil {
		return err
	}
	go func() {
		err := cg.Consume(context.Background(), []string{topic}, saramax.NewHandler[MsgEvent](ws.consume))
		if err != nil {
			log.Println("退出监听消息循环", err)
		}
	}()
	return nil
}
func (ws *WsGateway) consume(msg *sarama.ConsumerMessage, t MsgEvent) error {
	fmt.Println(t)
	conn, ok := ws.conns.Load(t.Receiver)
	if !ok {
		log.Printf("当前节点上没有这个用户:%v，直接返回\n", t.Receiver)
		return nil
	}
	return conn.Send(t.Msg)
}

// Conn 稍微做一个封装
type Conn struct {
	*websocket.Conn
}

func (c *Conn) Send(msg Message) error {
	val, err := json.Marshal(msg)
	if err != nil {
		return err
	}
	return c.WriteMessage(websocket.TextMessage, val)
}

type Message struct {
	// 发过来的消息的序列号
	// 用于前后端关联消息
	Seq string
	// 标识不同的消息类型
	Type    string
	Content string
	// 聊天 ID，注意，正常来说这里不是记录目标用户 ID
	// 而是记录代表了这个聊天的 ID
	Cid int64
}
